package org.whl.pollution.analysis
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{Row, SparkSession}
import org.whl.util.spark.SparkUtil
/**
 * @author 王浩霖
 * @version 1.0.0 2024/12/30 8:58
 */
object HumAirquality {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkUtil()

    // 从Hive数据库加载数据
    val df = spark.read.table("wanghaolin.whl_pollution")

    // 定义湿度分段
    val humidityRanges = Array((30, 50), (50, 70), (70, 90), (90, 110))

    // 初始化结果DataFrame的列
    var resultDF = spark.createDataFrame(Seq(
      ("30~50", 0, 0, 0, 0, 0),
      ("50~70", 0, 0, 0, 0, 0),
      ("70~90", 0, 0, 0, 0, 0),
      ("90~110", 0, 0, 0, 0, 0)
    )).toDF("Humidity", "PM25", "PM10", "SO2", "NO2", "CO")

    // 调用隐式转换
    import spark.implicits._

    // 对每个湿度范围进行过滤和计数
    humidityRanges.foreach { range =>
      val filteredDF = df.filter($"Humidity" >= range._1 && $"Humidity" < lit(range._2))

      val pm25Count = filteredDF.filter($"PM25" > 75).count()
      val pm10Count = filteredDF.filter($"PM10" > 150).count()
      val so2Count = filteredDF.filter($"SO2" > 40).count()
      val no2Count = filteredDF.filter($"NO2" > 40).count()
      val coCount = filteredDF.filter($"CO" > 3).count()

      // 更新结果DataFrame
      val index = range match {
        case (30, 50) => 0
        case (50, 70) => 1
        case (70, 90) => 2
        case (90, 110) => 3
      }

      resultDF = resultDF.withColumn("PM25", resultDF("PM25") + lit(pm25Count.toInt))
        .withColumn("PM10", resultDF("PM10") + lit(pm10Count.toInt))
        .withColumn("SO2", resultDF("SO2") + lit(so2Count.toInt))
        .withColumn("NO2", resultDF("NO2") + lit(no2Count.toInt))
        .withColumn("CO", resultDF("CO") + lit(coCount.toInt))
    }

    // 将结果DataFrame注册为临时视图，以便写入Hive表
    resultDF.createOrReplaceTempView("humidity_pollution_view")

    // 将结果写入Hive表
    spark.sql("CREATE TABLE IF NOT EXISTS wanghaolin.whl_hum_air AS SELECT * FROM humidity_pollution_view").collect()

    // 停止SparkSession
    spark.stop()
  }

}
